package com.shujia.spark.streaming

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
object Demo5ReadKafka {

  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setMaster("local[2]")
      .setAppName("readKafka")

    /**
      * 创建streaming上下文对象，指定batch的间隔时间，多久计算一次
      */

    val ssc: StreamingContext = new StreamingContext(conf,Durations.seconds(5))

    ssc.checkpoint("data/checkpoint")

    //读取kafka数据

    //kafka链接参数

    /**
      * earliest
      * 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
      * latest
      * 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
      * none
      * topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
      *
      */

    val kafkaParams: Map[String,Object] = Map[String,Object](
      "bootstrap.servers" ->"master:9092,node1:9092,node2:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "gId3",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> "false"
    )

    //topic列表
    val topics = Array("test_topic1")

    /**
      * createDirectStream : 主动拉取数据
      */

    val linesDS: InputDStream[ConsumerRecord[String, String]] =  KafkaUtils.createDirectStream[String,String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    /**
      * kafka是一个key,value格式的，默认key为null，一般用不上
      */

    linesDS
      .map(record => (record.key(),record.value()))
      .map(_._2)
      .flatMap(_.split(","))
      .map((_,1))
      .updateStateByKey((seq: Seq[Int],option: Option[Int]) =>Some(seq.sum + option.getOrElse(0)))
      .print()


    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
